package com.wq.concurrency.aqs;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Author: wangqiang20995
 * @Date:2018/10/28
 * @Description:
 * @Resource:
 */
public class SimpleBlockingQueue {

    final Lock lock = new ReentrantLock(true);
    final Condition notFull = lock.newCondition();
    final Condition notEmpty = lock.newCondition();

    Object[] queue;
    int putStr, takeStr, count;//putStr代表当前队列里面，第一个空着的位子，因为队列是数组实现，所以putStr必然是[0,size)之间，也就是数组的下标循环

    public SimpleBlockingQueue(int size) {
        queue = new Object[size];
    }

    public void put(Object item) throws InterruptedException {
        lock.lock();
        try {
            while (this.count == queue.length) {
                System.out.println("full queue and current [" + Thread.currentThread().getName() + "] wait until queue not full");
                notFull.await();
            }

            this.queue[putStr] = item;
            //判断自增后的下标是否到最后一个了，如果是的话回到数组的第一个下标
            if (++this.putStr == this.queue.length) {
                System.out.println("beyond queue item and come back to the first one");
                this.putStr = 0;
            }

            ++count;
            this.notEmpty.signal();//唤醒获取的东西的阻塞线程，可以从queue中获取东西了
        } finally {
            lock.unlock();
        }
    }

    public Object take(int index) throws InterruptedException {
        lock.lock();
        try {
            while (this.count == 0) {
                System.out.println("empty queue and current [" + Thread.currentThread().getName() + "] wait until queue not empty");
                this.notEmpty.await();
            }

            Object object = this.queue[takeStr];
            //代表已经拿到了队列中下标为size-1的元素，再往后的话就需要从第一个元素获取了
            if (++this.takeStr == this.queue.length) {
                this.takeStr = 0;
            }

            --count;
            this.notFull.signal();//唤醒生产东西的阻塞线程，可以继续往queue里面塞东西了
            return object;
        } finally {
            lock.unlock();
        }
    }

    public static void main(String args[]) {
        SimpleBlockingQueue simpleBlockingQueue = new SimpleBlockingQueue(5);
        int producers = 9, consumers = 10;
        for (int i = 0; i < producers; i++) {
            int finalI = i;
            Thread thread = new Thread(() -> {
                try {
                    Integer integer = new Integer(finalI + 1);
                    simpleBlockingQueue.put(integer);
                    System.out.println("[" + Thread.currentThread().getName() + "] put item --> [" + integer + "]");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
            thread.setName("producer-" + (i + 1));
            thread.start();
        }

        AtomicInteger queueIndex = new AtomicInteger();
        for (int i = 0; i < consumers; i++) {
            int finalI = i;
            Thread consumer = new Thread(() -> {
                try {
                    Integer integer = (Integer) simpleBlockingQueue.take(queueIndex.get());
                    System.out.println("[" + Thread.currentThread().getName() + "] take item --> [" + integer + "]");
                    queueIndex.getAndIncrement();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
            consumer.setName("consumer-" + (i + 1));
            consumer.start();
        }
    }
}
